/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.queue.impl;

import java.util.concurrent.ThreadFactory;

import org.sabayframework.channel.Channel;

/**
 * TODO document class 
 * 
 * @author Alex Antonau
 * */
public class DistributedTaskStealingQueue extends BaseExecutionQueue {
	
	private Worker first;
	private final ThreadLocal<Worker> current=new ThreadLocal<Worker>();
	private final int maxPollLatencyMillis;
	
	public DistributedTaskStealingQueue() {
		this(Runtime.getRuntime().availableProcessors());
	}
	
	public DistributedTaskStealingQueue(int threadNumber) {
		this(new DefaultThreadFactory(), threadNumber, 100);
	}
	
	public DistributedTaskStealingQueue(ThreadFactory threadFactory, int threadNumber, int maxPollLatencyMillis) {
		super(threadFactory);
		if(threadNumber < 1) throw new IllegalArgumentException(threadNumber+" < 1");
		this.maxPollLatencyMillis = maxPollLatencyMillis;
		start(threadNumber);
	}

	class Worker implements Runnable{
		
		private final ArrayDataQueue<ActivityContextImpl<?,?>> queue;
		private final Monitor monitor;
		private volatile ActivityContextImpl<?,?> running;
		Worker next;
		Worker prev;
		
		public Worker() {
			if(first == null){
				first = prev = next = this;
			} else {
				final Worker last = first.prev;
				next = last.next;
				prev = last;
				last.next = this;
				first.prev = this;
			}
			monitor = new Monitor();
			queue = new ArrayDataQueue<ActivityContextImpl<?,?>>(1000);
		}
		
		@Override
		public void run() {
			current.set(this);
			while(active){
				try{
					final ActivityContextImpl<?,?> a = nextReady();
					running=a;
					runActivity(a, false);
					returnActivity(a);
					running=null;
				} 
				catch (InterruptedException e) {}
				catch (Throwable t) {
					lastError = t;
				}
			}
			current.remove();
		}
		private void debriefActivity(ActivityContextImpl<?,?> a){
			a.afterRemoving(DistributedTaskStealingQueue.this);
		}
		
		private void returnActivity(final ActivityContextImpl<?, ?> a) {
			if(!a.isDone()) queue.add(a);
			else debriefActivity(a);
		}
		private ActivityContextImpl<?, ?> pollQueue(ArrayDataQueue<ActivityContextImpl<?, ?>> queue, int size){
			//loop through the queue and look for ready activities
			while(size-->0){
				final ActivityContextImpl<?,?> a = queue.poll();
				if(a == null) continue;
				if(a.isReady()) {					
					return a;
				}
				if(!a.isDone()) queue.add(a);
				else debriefActivity(a);
			};
			return null;
		}
		
		private ActivityContextImpl<?, ?> nextReady() throws InterruptedException {
			while(active){
				final int qsize = queue.size();
				ActivityContextImpl<?,?> a = pollQueue(queue, qsize);
				if(a!=null) return a;			
				Worker worker = prev;
				while(worker!=this){
					a = worker.running;
					if(a!=null){
						final int prevQsize = worker.queue.size();
						//balance activities if needed
						final int steal = (prevQsize - qsize) / 2;
						if(steal > 0){
							//steal activities from tail of a queue
							queue.addFromTail(worker.queue, steal);
							continue;
						}
						//steal activity from head of a queue
						a = pollQueue(worker.queue, prevQsize);
						if(a!=null) return a;
					}
					worker = worker.prev;
				}
				monitor.lock();
				try{
					if(queue.size() == qsize) monitor.waitChanged(maxPollLatencyMillis);
				} finally {
					monitor.unlock();
				}
			}
			throw new InterruptedException();
		}
		
		void add(ActivityContextImpl<?,?> ctx){
			queue.add(ctx);
			if(running == null)monitor.changed();
		}
		
		boolean isEmpty(){
			return queue.isEmpty();
		}
		
		int size(){
			return queue.size();
		}
		boolean channelChanged(Channel<?> input){
//			Iterator<ActivityContextImpl<?, ?>> it = queue.iterator();
//			while(it.hasNext()){
//				ActivityContextImpl<?,?> ctx = it.next();
//				if(ctx.getInput().equals(input)){
//					if(running == null) monitor.changed();
//					return true;
//				}
//			}
			return false;
		}
	}
	
	@Override
	public void channelChanged(Channel<?> input) {
		Worker w = first;
		do{
			w.channelChanged(input);
			w = w.next;
		}while(w!=first);
		
	}

	@Override
	public Runnable createWorker(int workerNo) {
		return new Worker();
	}

	@Override
	protected <I, O> void assignToWorker(ActivityContextImpl<I, O> ctx){
		Worker selected = current.get();
		if(selected==null) selected = first;
		selected.add(ctx);
	}

	@Override
	public int size() {
		int count =0;
		Worker w = first;
		do{
			count+=w.size();
			w = w.next;
		}while(w!=first);
		return count;
	}

	@Override
	public boolean isEmpty() {
		Worker w =first;
		do{
			if(!w.isEmpty()) return false;
			w = w.next;
		}while(w!=first);
		return true;
	}

}
